package outbound;



import exception.AuthException;
import filter.HeaderHttpResponseFilter;
import filter.HttpRequestFilter;
import filter.HttpResponseFilter;
import generator.ResponseGenerator;
import io.micrometer.core.instrument.util.NamedThreadFactory;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpUtil;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;

import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import property.RouteProperty;
import router.HttpEndpointRouter;
import router.RandomHttpEndpointRouter;

import java.util.concurrent.*;

import static io.netty.handler.codec.http.HttpResponseStatus.*;

public class HttpOutboundHandler {


    private CloseableHttpAsyncClient httpclient;
    private ExecutorService proxyService;
    private RouteProperty routeProperty;

    HttpResponseFilter filter = new HeaderHttpResponseFilter();

    HttpEndpointRouter router = new RandomHttpEndpointRouter();

    public HttpOutboundHandler(RouteProperty routeProperty) {

        this.routeProperty = routeProperty;

        int cores = Runtime.getRuntime().availableProcessors();
        long keepAliveTime = 1000;
        int queueSize = 2048;
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();//.DiscardPolicy();
        proxyService = new ThreadPoolExecutor(cores, cores,
                keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize),
                new NamedThreadFactory("proxyService"), handler);

        IOReactorConfig ioConfig = IOReactorConfig.custom()
                .setConnectTimeout(1000)
                .setSoTimeout(1000)
                .setIoThreadCount(cores)
                .setRcvBufSize(32 * 1024)
                .build();
        httpclient = HttpAsyncClients.custom().setMaxConnTotal(40)
                .setMaxConnPerRoute(8)
                .setDefaultIOReactorConfig(ioConfig)
                .setKeepAliveStrategy((response, context) -> 6000)
                .build();
        httpclient.start();
    }
    public void handle(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, HttpRequestFilter filter) {
        String backendUrl="";
        //路由获取host
        backendUrl=router.route(routeProperty,fullRequest.uri());
        final String url = backendUrl + fullRequest.uri();
        //对请求头进行自定义，通过设置"yourname"来使某些服务通过验证
        filter.filter(fullRequest, ctx);
        proxyService.submit(() -> fetchGet(fullRequest, ctx, url));
    }

    private void fetchGet(final FullHttpRequest inbound, final ChannelHandlerContext ctx, final String url) {
        final HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
        httpGet.setHeader("name", inbound.headers().get("name"));

        httpclient.execute(httpGet, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(final HttpResponse endpointResponse) {
                try {
                    handleResponse(inbound, ctx, endpointResponse);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {

                }
            }
            @Override
            public void failed(final Exception ex) {
                httpGet.abort();
                ex.printStackTrace();
            }

            @Override
            public void cancelled() {
                httpGet.abort();
            }
        });
    }

    private void handleResponse(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, final HttpResponse endpointResponse) throws Exception {
        FullHttpResponse response = null;
        try {
            //优先判断响应是否出现401
            filter.filter(endpointResponse);
            byte[] body = EntityUtils.toByteArray(endpointResponse.getEntity());
            response=ResponseGenerator.getResponse(OK,Unpooled.wrappedBuffer(body),endpointResponse);
        }catch (AuthException authException)
        {
            authException.printStackTrace();
            response=ResponseGenerator.getResponse(UNAUTHORIZED,Unpooled.wrappedBuffer("UNAUTHORIZED".getBytes()),endpointResponse);
        }
        catch (Exception e) {
            e.printStackTrace();
            response=ResponseGenerator.getResponse(NO_CONTENT,Unpooled.wrappedBuffer("NO_CONTENT".getBytes()),endpointResponse);
        } finally {
            if (fullRequest != null) {
                if (!HttpUtil.isKeepAlive(fullRequest)) {
                    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
                } else {
                    ctx.write(response);
                }
            }
            ctx.flush();
        }

    }
}
